package com.xxxx;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;

public class Hello08FlinkGlobalWindow {
    public static void main(String[] args) throws Exception {
        //1.初始化环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        //2.读取数据源
        DataStreamSource<String> source = environment.socketTextStream("192.168.88.101", 18880);
        source.map(word -> {
            System.err.println(word + "--" + System.currentTimeMillis());
            return Tuple2.of(word, 1);
        }).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy(0)
                .windowAll(GlobalWindows.create())  // 使用全局窗口
                .trigger(CountTrigger.of(100))  // 每满100个元素触发一次计算
                .sum(1) // 聚合
                .print("张三:");  // 打印

        //3.需要手动触发流式计算的执行效果
        environment.execute("Hello03FlinkSessionWindow" + System.currentTimeMillis());
    }
}